// using System;
// using System.Threading.Channels;
// using System.Threading.Tasks;
// using Grpc.Core;
// using Microsoft.Extensions.Logging;
// using Realtime;

// namespace RealTimeChat.Services
// {
//     public class ChatServiceImpl : ChatService.ChatServiceBase
//     {
//         private readonly ILogger<ChatServiceImpl> _logger;
//         public ChatServiceImpl(ILogger<ChatServiceImpl> logger) => _logger = logger;

//         public override async Task Chat(
//             IAsyncStreamReader<ChatMessage> requestStream,
//             IServerStreamWriter<ChatMessage> responseStream,
//             ServerCallContext context)
//         {
//             var ct = context.CancellationToken;
//             var channel = Channel.CreateBounded<ChatMessage>(new BoundedChannelOptions(1000)
//             {
//                 FullMode = BoundedChannelFullMode.Wait
//             });

//             // 生产者：客户端 → Channel
//             var readerTask = Task.Run(async () =>
//             {
//                 await foreach (var msg in requestStream.ReadAllAsync(ct))
//                     await channel.Writer.WriteAsync(msg, ct);
//                 channel.Writer.Complete();
//             }, ct);

//             // 消费者：Channel → 客户端
//             var writerTask = Task.Run(async () =>
//             {
//                 await foreach (var msg in channel.Reader.ReadAllAsync(ct))
//                 {
//                     var time = DateTimeOffset.FromUnixTimeSeconds(msg.Timestamp)
//                                            .ToLocalTime().ToString("o");
//                     _logger.LogInformation("[{Time}] {User}: {Message}", time, msg.User, msg.Message);
//                     try
//                     {
//                         await responseStream.WriteAsync(msg, ct);
//                     }
//                     catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
//                     {
//                         _logger.LogWarning("Client cancelled"); break;
//                     }
//                     catch (RpcException ex) when (ex.StatusCode == StatusCode.DeadlineExceeded)
//                     {
//                         _logger.LogWarning("Write timed out"); break;
//                     }
//                     catch
//                     {
//                         _logger.LogError("Write failed"); break;
//                     }
//                 }
//             }, ct);

//             await Task.WhenAll(readerTask, writerTask);
//         }
//     }
// }

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Realtime;

namespace RealTimeChat.Services
{
    public class ChatServiceImpl : ChatService.ChatServiceBase
    {
        private readonly ILogger<ChatServiceImpl> _logger;

        // 存放所有活跃连接的 responseStream
        private static readonly ConcurrentDictionary<Guid, IServerStreamWriter<ChatMessage>> _peers
            = new();

        public ChatServiceImpl(ILogger<ChatServiceImpl> logger) => _logger = logger;

        public override async Task Chat(
            IAsyncStreamReader<ChatMessage> requestStream,
            IServerStreamWriter<ChatMessage> responseStream,
            ServerCallContext context)
        {
            // 为每个连接分配一个唯一 ID
            var connectionId = Guid.NewGuid();
            _peers[connectionId] = responseStream;

            try
            {
                await foreach (var msg in requestStream.ReadAllAsync(context.CancellationToken))
                {
                    var time = DateTimeOffset
                        .FromUnixTimeSeconds(msg.Timestamp)
                        .ToLocalTime()
                        .ToString("o");
                    _logger.LogInformation("[{Time}] {User}: {Message}", time, msg.User, msg.Message);

                    // 广播给除自己以外的所有客户端
                    foreach (var kv in _peers)
                    {
                        if (kv.Key == connectionId)
                        {
                            // 跳过自己
                            continue;
                        }

                        try
                        {
                            await kv.Value.WriteAsync(msg);
                        }
                        catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
                        {
                            // 忽略已断开的连接
                        }
                        catch (Exception ex)
                        {
                            _logger.LogError(ex, "Failed to broadcast message to peer {PeerId}", kv.Key);
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // 客户端主动取消流，直接返回
            }
            finally
            {
                // 连接结束时移除
                _peers.TryRemove(connectionId, out _);
            }
        }
    }
}

